package edu.zut.cs.network.socket;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;

import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.util.ByteSequence;
import edu.zut.cs.network.base.BaseMainCase;
import edu.zut.cs.network.entity.Book;
import edu.zut.cs.network.iountil.WriteToCSV;

/**
 * 消息生产者-消息发布者（多线程发送消息）
 *
 */
public class JMSConsumerThread extends BaseMainCase{

    private static final String USERNAME="yrl"; // 默认的连接用户名
    private static final String PASSWORD="yrl"; // 默认的连接密码
    private static final String BROKEURL="tcp://192.168.57.138:61616"; // 默认的连接地址

    ConnectionFactory connectionFactory=null; // 连接工厂
    private Connection connection = null;
    private Session session = null;
    private Destination destination=null; // 消息的目的地
    public JMSConsumerThread(){
        // 实例化连接工厂
    	connectionFactory=new ActiveMQConnectionFactory(JMSConsumerThread.USERNAME, JMSConsumerThread.PASSWORD, JMSConsumerThread.BROKEURL);        try {
            connection=connectionFactory.createConnection(); // 通过连接工厂获取连接
            connection.start();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
    public void consumer(){
        MessageConsumer messageConsumer; // 消息的消费者

        try {
            session=connection.createSession(Boolean.FALSE, Session.AUTO_ACKNOWLEDGE); // 创建Session
            destination=session.createQueue("sqlqueue1");
            messageConsumer=session.createConsumer(destination); // 创建消息消费者
            messageConsumer.setMessageListener(new MessageListener() {

    			@Override
    			public void onMessage(Message message) {
    				// 打印结果
    				
    				BytesMessage bm = (BytesMessage) message;
    				
    				Field field = null;
    				Class<?> clazz = bm.getClass();
    				for (; clazz != Object.class; clazz = clazz.getSuperclass()) {
    					try {
    						field = clazz.getDeclaredField("content");
    					} catch (Exception e) {
    						// 这里甚么都不能抛出去。
    						// 如果这里的异常打印或者往外抛，则就不会进入
    					}
    				}
    				field.setAccessible(true);
    				byte[] by = null;
    				try {
    					ByteSequence b = (ByteSequence) field.get(bm);
    					by = b.getData();
    				} catch (IllegalArgumentException e1) {
    					e1.printStackTrace();
    				} catch (IllegalAccessException e1) {
    					e1.printStackTrace();
    				}
    				List<Book> l = unserializeForList(by);
    				new WriteToCSV(l);//写入文件
    				logger.info("size is:"+l.size()+"\n数据已存于<D:/bookqueqe.csv>");
    			}

    		});
         // 等待接收消息
    		System.in.read();
    		// 关闭资源
    		messageConsumer.close();
    		session.close();
    		connection.close();

        } catch (JMSException e) {
            e.printStackTrace();
        } catch (IOException e) {
			e.printStackTrace();
		}
    }
	public static <T> List<T> unserializeForList(byte[] in) {
		List<T> list = new ArrayList<T>();
		ByteArrayInputStream bis = null;
		ObjectInputStream is = null;
		try {
			if (in != null) {
				bis = new ByteArrayInputStream(in);
				is = new ObjectInputStream(bis);
				while (true) {
					@SuppressWarnings("unchecked")
					T obj = (T) is.readObject();
					if (obj == null) {
						break;
					} else {
						list.add(obj);
					}
				}
				is.close();
				bis.close();
			}
		} catch (IOException e) {
			logger.warn("Caught IOException decoding %d bytes of data", in == null ? 0 : in.length, e);
		} catch (ClassNotFoundException e) {
			logger.warn("Caught CNFE decoding %d bytes of data", in == null ? 0 : in.length, e);
		}
		return list;
	}

}
